using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Configuration;
using System.Diagnostics;
using System.IO;
using System.Net;
using System.Net.Mail;
using System.Reflection;
using System.Text;
using System.Threading;
using System.Timers;
using System.Xml;
using Common.Utilities;
using Common.Utilities.Dispatcher;
using Common.Utilities.DLR;
using Workflows.Components.DALC;
using Timer=System.Timers.Timer;

namespace Workflows.Components
{
    /// <summary></summary>
    public class JobDispatcherService
    {
        #region settings
        private Mutex _SyncObj;
        private bool _IsActive;
        private int RETRIEVAL_INTERVAL = 1000*60;   // 1 MIN
        private int DISPATCHER_INTERVAL = 1000*5;  // 5 SEC
        private int JOB_POOL = 4;
        private string EVENT_NAME = "JOB DISPATCHER";
        private string EVENT_SORUCE = "Workflow";

        // email
        private string _SMTPServerName;
        private string _FromEmailAddress;
        private string _UserName;
        private string _Password;
        private bool _RequireAuthentication;
        private bool _EnableSSL;
        #endregion

        #region fields
        private EventLog _Log;
        private Timer _RetrievalTimer;
        private Timer _DispatcherTimer;
        private Dictionary<Guid,Job> _Jobs;
        private Dictionary<Guid, Guid> _ComponentJobMaps;       // component id->job id
        private Dictionary<string, WorkflowComponentBase> _HandlableComponents;       // componentid->component
        private Dictionary<Guid, IWorker> _JobWorkMaps;     // job id -> worker 
        private Dictionary<Type, Type> _ComponentWorkerType;    // componentType->workertype
        private Dictionary<string, Type> _InstructionTypes;      // map typefullname to instruction type
        private Dictionary<string, Type> _OutputTypes;            // map typefullname to output type
        private int _OngoingWorkerCount;
        #endregion

        #region init
        /// <summary></summary>
        public JobDispatcherService()
        {
            this._SyncObj=new Mutex(false, "JobQueue");

            this.EVENT_SORUCE = ConfigurationManager.AppSettings.Get("EventLogSource");
            this.EVENT_NAME = ConfigurationManager.AppSettings.Get("EventLogName");
            this.RETRIEVAL_INTERVAL = int.Parse(ConfigurationManager.AppSettings.Get("JobRetrievalInterval"));
            this.DISPATCHER_INTERVAL = int.Parse(ConfigurationManager.AppSettings.Get("JobDispatcherInterval"));
            this.JOB_POOL = int.Parse(ConfigurationManager.AppSettings.Get("JobPool"));
            this._Jobs= new Dictionary<Guid, Job>();
            this._HandlableComponents = new Dictionary<string, WorkflowComponentBase>();
            this._ComponentJobMaps=new Dictionary<Guid, Guid>();
            this._ComponentWorkerType = new Dictionary<Type, Type>();
            this._JobWorkMaps=new Dictionary<Guid, IWorker>();
            this._InstructionTypes = new Dictionary<string, Type>();
            this._OutputTypes = new Dictionary<string, Type>();
            this._OngoingWorkerCount = 0;
            
            this.RegisterEventLog();

            this._SMTPServerName = ConfigurationManager.AppSettings.Get("SmtpServer");
            this._FromEmailAddress = ConfigurationManager.AppSettings.Get("FromEmailAddress");
            this._UserName = ConfigurationManager.AppSettings.Get("EmailUserName");
            this._Password = ConfigurationManager.AppSettings.Get("EmailPassword");
            this._RequireAuthentication = bool.Parse(ConfigurationManager.AppSettings.Get("RequireAuthentication"));
            this._EnableSSL = bool.Parse(ConfigurationManager.AppSettings.Get("EnableSSL"));
        }

        private void RegisterEventLog()
        {
            if(!EventLog.SourceExists(this.EVENT_SORUCE))
            {
                EventLog.CreateEventSource(this.EVENT_SORUCE,this.EVENT_NAME);
            }
            this._Log=new EventLog(this.EVENT_NAME,".",this.EVENT_SORUCE);
            this._Log.MaximumKilobytes = 4096;
            this._Log.ModifyOverflowPolicy(OverflowAction.OverwriteAsNeeded, 7);
        }

        private void RegisterHandlableComponents()
        {
            this._HandlableComponents = new Dictionary<string, WorkflowComponentBase>();
            this._ComponentWorkerType=new Dictionary<Type, Type>();
            string dllFolderPath = Path.GetDirectoryName(Assembly.GetExecutingAssembly().Location);
            DirectoryInfo binFolderInfno=new DirectoryInfo(dllFolderPath);
            FileInfo[] libInfos = binFolderInfno.GetFiles("*.dll", SearchOption.TopDirectoryOnly);
            foreach(FileInfo libFileInfo in libInfos)
            {
                try
                {
                    Assembly libAssembly = Assembly.LoadFile(libFileInfo.FullName);
                    Type[] libTypes = libAssembly.GetTypes();
                    foreach (Type type in libTypes)
                    {
                        if (type.IsSubclassOf(typeof (WorkflowComponentBase)) && (!type.IsAbstract))
                        {
                            WorkflowComponentBase wfComponent = (WorkflowComponentBase) Activator.CreateInstance(type);
                            this._HandlableComponents.Add(type.FullName, wfComponent);
                        }

                        if (type.GetInterface(typeof (IWorker).FullName) != null && (!type.IsAbstract))
                        {
                            IWorker worker = (IWorker) Activator.CreateInstance(type);
                            if (!this._ComponentWorkerType.ContainsKey(worker.ComponentType))
                                this._ComponentWorkerType.Add(worker.ComponentType, type);
                        }
                        if (type.IsSubclassOf(typeof (InstructionBase)) &&
                            (!this._InstructionTypes.ContainsKey(type.FullName)))
                        {
                            this._InstructionTypes.Add(type.FullName, type);
                        }
                        if (type.IsSubclassOf(typeof (ResultBase)) && (!this._OutputTypes.ContainsKey(type.FullName)))
                        {
                            this._OutputTypes.Add(type.FullName, type);
                        }
                    }
                }
                catch (Exception ex)
                {
                    Console.WriteLine("Unable to load assembly: " + ex.ToString());
                }
            }
        }
        #endregion

        #region start
        /// <summary></summary>
        public void Start()
        {
            this._SyncObj.WaitOne();
            this._OngoingWorkerCount = 0;
            this._Jobs.Clear();
            this.RegisterHandlableComponents();
            this.StartRetrieval();
            this.StartDispatcher();
            this._IsActive = true;
            this._SyncObj.ReleaseMutex();
        }

        private void StartRetrieval()
        {
            if (this._RetrievalTimer != null)
                this._RetrievalTimer.Dispose();
            GC.Collect();
            GC.WaitForFullGCComplete();
            this._RetrievalTimer=new Timer(this.RETRIEVAL_INTERVAL);
            this._RetrievalTimer.Enabled = true;
            this._RetrievalTimer.Elapsed += new ElapsedEventHandler(RetrieveJobs);
            this._RetrievalTimer.Start();
        }

        private void StartDispatcher()
        {
            this.JOB_POOL = int.Parse(ConfigurationManager.AppSettings.Get("JobPool"));
            if (this._DispatcherTimer != null)
                this._DispatcherTimer.Dispose();
            this._DispatcherTimer = null;
            GC.Collect();
            GC.WaitForFullGCComplete();
            this._DispatcherTimer=new Timer(DISPATCHER_INTERVAL);
            this._DispatcherTimer.Enabled = true;
            this._DispatcherTimer.Elapsed += new ElapsedEventHandler(DispatchJobs);
            this._DispatcherTimer.Start();
        }

        /// <summary></summary>
        public bool IsActive
        {
            get
            {
                return this._IsActive;
            }
        }
        #endregion

        #region stop
        /// <summary></summary>
        public void Stop()
        {
            this._SyncObj.WaitOne();
            this._OngoingWorkerCount = 0;
            this._Jobs.Clear();
            this.StopRetrieval();
            this.StopDispatcher();
            this._IsActive = false;
            this._SyncObj.ReleaseMutex();
        }

        /// <summary></summary>
        public void StopRetrieval()
        {
            this._RetrievalTimer.Stop();
        }

        /// <summary></summary>
        public void StopDispatcher()
        {
            this._DispatcherTimer.Stop();
        }
        #endregion

        #region retrieval
        void RetrieveJobs(object sender, ElapsedEventArgs e)
        {
            this._SyncObj.WaitOne();
            this.StopRetrieval();
            try
            {
                this._SyncObj.WaitOne();
                List<string> componentTypeNames = new List<string>();
                foreach (string componentTypeName in this._HandlableComponents.Keys)
                {
                    componentTypeNames.Add(componentTypeName);
                }
                List<Job> newJobList = JobDALC.CheckoutNewJobs(componentTypeNames);
                if (newJobList != null && newJobList.Count > 0)
                {
                    for (int i = 0; i < newJobList.Count; i++)
                    {
                        Job newJob = newJobList[i];
                        if (!this._Jobs.ContainsKey(newJob.JobID))
                        {
                            newJob.ReceiveTime = DateTime.Now;
                            this._Jobs.Add(newJob.JobID, newJob);
                            JobDALC.SaveJob(newJob);
                        }
                    }
                    this.LogInfo("Retrieved " + newJobList.Count + " jobs.");
                }
            }
            catch (Exception ex)
            {
                this.LogErrorMessage(ex.ToString());
                this.LogErrorMessage(ex.StackTrace);
            }
            finally
            {
                this._SyncObj.ReleaseMutex();
            }

            this.StartRetrieval();
            this._SyncObj.ReleaseMutex();
        }
        #endregion

        #region dispatcher 
        void DispatchJobs(object sender, ElapsedEventArgs e)
        {
            this._SyncObj.WaitOne();
            this.StopDispatcher();
            try
            {
                this._SyncObj.WaitOne();
                if (this._OngoingWorkerCount < this.JOB_POOL && this._Jobs.Count > 0)
                {
                    Guid[] jobIDs = new Guid[this._Jobs.Count];
                    this._Jobs.Keys.CopyTo(jobIDs, 0);
                    foreach (Guid jobID in jobIDs)
                    {
                        Job nextJob = this._Jobs[jobID];
                        if(nextJob.SubmissionTime <DateTime.Now && nextJob.SubmissionTime > TimeUtil.DbMinTime)
                        {
                            // already submitted not not finished
                            continue;
                        }
                        if (nextJob.StartTime == TimeUtil.DbMinTime)
                        {
                            this.LogInfo("Trying to dispatch next queued jobs...");
                            IWorker worker = this.CreateWorkerAndSetInstruction(nextJob);
                            if (worker != null)
                            {
                                this._Jobs[nextJob.JobID] = nextJob;
                                if (this._JobWorkMaps.ContainsKey(nextJob.JobID))
                                {
                                    this._JobWorkMaps[nextJob.JobID] = worker;
                                }
                                else
                                {
                                    this._JobWorkMaps.Add(nextJob.JobID, worker);
                                }
                                worker.DoWorkAsync();

                                this._OngoingWorkerCount += 1;
                            }
                            else
                            {
                                nextJob.SubmissionTime = TimeUtil.DbMinTime;
                                this._Jobs[nextJob.JobID] = nextJob;
                            }
                            break;
                        }
                    }

                }
            }
            catch (Exception ex)
            {
                this.LogErrorMessage(ex.ToString());
                this.LogErrorMessage(ex.StackTrace);
            }
            finally
            {
                this._SyncObj.ReleaseMutex();
            }

            this.StartDispatcher();
            this._SyncObj.ReleaseMutex();
            
        }

        private IWorker CreateWorkerAndSetInstruction(Job job)
        {
            job.SubmissionTime = DateTime.Now;
            JobDALC.SaveJob(job);

            WorkflowComponentBase associatedComponent = this._HandlableComponents[job.ComponentTypeName];
            Type workerType = this._ComponentWorkerType[associatedComponent.GetType()];
            IWorker workerInstance = (IWorker) Activator.CreateInstance(workerType);
            workerInstance.WorkerCancelledEvent += new WorkerCancelled(workerInstance_WorkerCancelledEvent);
            workerInstance.WorkerErredEvent += new WorkerHasError(workerInstance_WorkerErredEvent);
            workerInstance.WorkerFinishedEvent += new WorkerFinished(workerInstance_WorkerFinishedEvent);
            workerInstance.WorkerStartedEvent += new WorkerStarted(workerInstance_WorkerStartedEvent);
            workerInstance.WorkerStatusChangedEvent += new WorkerStatusChanged(workerInstance_WorkerStatusChangedEvent);
            
            InstructionBase jobInstruction = this.ComputeInstructionForJob(job);
            Run run = RunDALC.GetRunByJob(job.JobID);
            if(run.Vars !=null && run.Vars.Count>0)
            {
                jobInstruction.BindToGlobalVariableValues(run.Vars);
            }
            WorkflowDesign wfDesign = WorkflowDesignDALC.GetWorkflowDesign(run.WorkflowID);
            WorkflowComponentBase wfComponent = WorkflowDesignDALC.GetWorkflowComponent(job.ComponentID);
            workerInstance.SetInstruction(job.JobID, jobInstruction, wfDesign.WorkflowName, wfComponent.ComponentName);
            // workerInstance.DoWork();
            return workerInstance;
        }

        void workerInstance_WorkerStatusChangedEvent(Guid ticketID, Type workerType, InstructionBase instructionBase,
            string workflowName, string componentName, string message, int percent)
        {
            this.LogInfo(string.Format("{0} | {1}({2}) status: {3}", workflowName, componentName, workerType.Name,
                                       message));
        }

        void workerInstance_WorkerStartedEvent(Guid ticketID, IWorker workerReference, Type workerType, InstructionBase instructionBase)
        {
            try
            {
                this._SyncObj.WaitOne();
                Job job = null;
                if (this._Jobs.ContainsKey(ticketID))
                {
                    job = this._Jobs[ticketID];
                    job.StartTime = DateTime.Now;
                }
                else
                {
                    job = JobDALC.GetJob(ticketID);
                    job.StartTime = DateTime.Now;
                }
                JobDALC.SaveJob(job);
                
                this.LogInfo(string.Format("Worker {0} started", workerType.Name));
                
                if(job.IsFirstJob)
                {
                    Run run = RunDALC.GetRunByJob(job.JobID);
                    run.StartTime = job.StartTime;
                    RunDALC.SaveRun(run);
                }
            }
            catch (Exception ex)
            {
                this.LogErrorMessage(ex.ToString());
                this.LogErrorMessage(ex.StackTrace);
            }
            finally
            {
                this._SyncObj.ReleaseMutex();
            }
           
        }

        void workerInstance_WorkerFinishedEvent(Guid ticketID, Type workerType, InstructionBase instructionBase, ResultBase resultBase)
        {
            try
            {
                this._SyncObj.WaitOne();
                string result = string.Empty;
                if (resultBase != null)
                {
                    XmlDocument xdoc = new XmlDocument();
                    xdoc.LoadXml("<Result></Result>");
                    XmlNode root = xdoc.DocumentElement;
                    resultBase.Serialize(ref xdoc, ref root);
                    JobOutputDALC.SaveJobOutput(resultBase);

                    result = xdoc.OuterXml;
                    if (result.Length > 255)
                    {
                        result = result.Substring(0, 255);
                    }
                }
                this.LogInfo(string.Format("Worker {0} finished with result {1}", workerType.Name, result));

                Job job = null;
                if (this._Jobs.ContainsKey(ticketID))
                {
                    job = this._Jobs[ticketID];
                }
                else
                {
                    job = JobDALC.GetJob(ticketID);
                }
                job.FinishTime = DateTime.Now;
                JobDALC.SaveJob(job);
                this._Jobs.Remove(ticketID);
                this._JobWorkMaps.Remove(ticketID);
                foreach (Guid componentID in this._ComponentJobMaps.Keys)
                {
                    Guid jobID = this._ComponentJobMaps[componentID];
                    if (jobID == ticketID)
                    {
                        this._ComponentJobMaps.Remove(componentID);
                    }
                }


                if (job.IsLastJob)
                {
                    Run run = RunDALC.GetRunByJob(job.JobID);
                    run.FinishTime = DateTime.Now;
                    RunDALC.SaveRun(run);
                }

                this._OngoingWorkerCount -= 1;
                this._OngoingWorkerCount = Math.Max(0, this._OngoingWorkerCount);
            }
            catch (Exception ex)
            {
                this.LogErrorMessage(ex.ToString());
                this.LogErrorMessage(ex.StackTrace);
            }
            finally
            {
                this._SyncObj.ReleaseMutex();
            }
        }

        void workerInstance_WorkerErredEvent(Guid ticketID, Type workerType, InstructionBase instructionBase, string error, bool isFatal)
        {
            try
            {
                this._SyncObj.WaitOne();
                this.LogErrorMessage(string.Format("Worker {0} has exception: {1}", workerType.Name, error));

                Job job = null;
                if (this._Jobs.ContainsKey(ticketID))
                {
                    job = this._Jobs[ticketID];
                }
                else
                {
                    job = JobDALC.GetJob(ticketID);
                }
                job.HasError = true;
                job.FinishTime = TimeUtil.DbMaxTime;
                JobDALC.SaveJob(job);

                if(isFatal)
                {
                    // decrement job queue
                    this._OngoingWorkerCount -= 1;
                    this._OngoingWorkerCount = Math.Max(0, this._OngoingWorkerCount);

                    // disable downstream jobs
                    List<Job> jobsDisabled = JobDALC.DisableJob(job.JobID);
                    foreach (Job jobDisabled in jobsDisabled)
                    {
                        this._Jobs.Remove(jobDisabled.JobID);
                        this._JobWorkMaps.Remove(jobDisabled.JobID);
                        foreach (Guid componentID in this._ComponentJobMaps.Keys)
                        {
                            Guid jobID = this._ComponentJobMaps[componentID];
                            if (jobID == jobDisabled.JobID)
                            {
                                this._ComponentJobMaps.Remove(componentID);
                            }
                        }
                    }

                    // send email notification
                    WorkflowDesign wfDesign = WorkflowDesignDALC.GetWorkflowDesignByRun(job.RunID);
                    if(!string.IsNullOrEmpty(wfDesign.FailureNotifyEmailAddress))
                    {
                        string[] toAddresses = wfDesign.FailureNotifyEmailAddress.Split(new char[] { ',', ';' });
                        string subject = string.Format("Error in workflow [{0}]", wfDesign.WorkflowName);
                        string body = string.Format("<b>{0}</b><br/><hr>{1}", job.ComponentTypeName, error);
                        this.SendNotification(this._SMTPServerName, this._EnableSSL, this._RequireAuthentication,
                                              this._UserName, this._Password, this._FromEmailAddress, toAddresses, null,
                                              subject, body);
                    }
                    
                }
            }
            catch (Exception ex)
            {
                this.LogErrorMessage(ex.ToString());
                this.LogErrorMessage(ex.StackTrace);
            }
            finally
            {
                this._SyncObj.ReleaseMutex();
            }
        }

        void workerInstance_WorkerCancelledEvent(Guid ticketID, Type workerType, InstructionBase instructionBase)
        {
            try
            {
                this._SyncObj.WaitOne();
                this.LogInfo(string.Format("Worker {0} is cancelled by user.", workerType.Name));

                if (this._Jobs.ContainsKey(ticketID))
                {
                    Job job = this._Jobs[ticketID];
                    job.Cancelled = true;
                    job.FinishTime = TimeUtil.DbMaxTime;
                    JobDALC.SaveJob(job);

                    //this._Jobs.Remove(ticketID);
                    //this._JobWorkMaps.Remove(ticketID);
                    //foreach (Guid componentID in this._ComponentJobMaps.Keys)
                    //{
                    //    Guid jobID = this._ComponentJobMaps[componentID];
                    //    if (jobID == ticketID)
                    //    {
                    //        this._ComponentJobMaps.Remove(componentID);
                    //    }
                    //}

                    List<Job> jobsDisabled = JobDALC.DisableJob(job.JobID);
                    foreach (Job jobDisabled in jobsDisabled)
                    {
                        this._Jobs.Remove(jobDisabled.JobID);
                        this._JobWorkMaps.Remove(jobDisabled.JobID);
                        foreach (Guid componentID in this._ComponentJobMaps.Keys)
                        {
                            Guid jobID = this._ComponentJobMaps[componentID];
                            if (jobID == jobDisabled.JobID)
                            {
                                this._ComponentJobMaps.Remove(componentID);
                            }
                        }
                    }
                }
                this._OngoingWorkerCount -= 1;
                this._OngoingWorkerCount = Math.Max(0, this._OngoingWorkerCount);
            }
            catch (Exception ex)
            {
                this.LogErrorMessage(ex.ToString());
                this.LogErrorMessage(ex.StackTrace);
            }
            finally
            {
                this._SyncObj.ReleaseMutex();
            }
        }
        #endregion

        #region data pipe
        /// <summary>
        /// Limitation:
        /// Instruction is instantiated from one type
        /// Unable to handle transformation, only assignment and type has to agree with each other
        /// </summary>
        /// <param name="job"></param>
        /// <returns></returns>
        private InstructionBase ComputeInstructionForJob(Job job)
        {
            // 1: get run id using job id (from workerinstanceid to workflowinstanceid)
            Run run = RunDALC.GetRunByJob(job.JobID);
            // 2: get workflow component by job id 
            
            // 3: get upstream connectors and components
            WorkflowDesign wfDesign = WorkflowDesignDALC.GetWorkflowDesignByRun(run.RunID);
            WorkflowComponentBase wfComponent = wfDesign.GetComponent(job.ComponentID);
            InstructionBase instruction = wfComponent.Instruction;
            List<Connector> upstreamConnectors = wfDesign.GetUpstreamConnectors(wfComponent.ComponentID);
            List<WorkflowComponentBase> upstreamComponents=new List<WorkflowComponentBase>();
            foreach(Connector wfConnector in upstreamConnectors)
            {
                WorkflowComponentBase upstreamComponent = wfDesign.GetComponent(wfConnector.FromComponentID);
                upstreamComponents.Add(upstreamComponent);
            }
            if(upstreamComponents.Count==0)
            {
                if(run.StartTime == TimeUtil.DbMinTime)
                {
                    run.StartTime = DateTime.Now;
                    RunDALC.SaveRun(run);
                }
            }
            // 4: get field mappers
            if(instruction !=null)
            {
                // Type instructionType = this._InstructionTypes[job.JobInstructionTypeName];
                instruction.OwnerWorkerInstanceID = job.JobID;
                instruction.StartTime = job.StartTime;
                instruction.SubmitTime = job.SubmissionTime;

                //if(instruction.GetType()==typeof(GlobalVariableGetterInstruction))
                //{
                //    // TODO: update instruction global variables with run 
                //    GlobalVariableGetterInstruction getterInstruction = (GlobalVariableGetterInstruction) instruction;
                //    getterInstruction.GlobalVariableNames=new List<string>();
                //    if(run.Vars !=null && run.Vars.Count>0)
                //    {
                        
                //    }
                //}
                // Dictionary<string, MemberInfo> writableMembers = TypeUtil.GetWritableMembers(instruction.GetType());
                Dictionary<string, Type> writableInstructionFields =
                    instruction.GetBindableMemberNames(new Type[] {typeof (InstructionBase)});
                
                if (upstreamComponents.Count > 0)
                {
                    foreach (WorkflowComponentBase upstreamComponent in upstreamComponents)
                    {
                        string outputXml = JobOutputDALC.GetJobOutput(run.RunID, upstreamComponent.ComponentID);
                        if(!string.IsNullOrEmpty(outputXml))
                        {
                            XmlDocument xdoc=new XmlDocument();
                            xdoc.LoadXml(outputXml);
                            if(xdoc.DocumentElement ==null)
                                throw new Exception("Invalid xdoc for job output");
                            XmlNode resultNode = xdoc.DocumentElement.ChildNodes[0];
                            string outputTypeName = resultNode.Attributes.GetNamedItem("Type").Value;
                            if(this._OutputTypes.ContainsKey(outputTypeName))
                            {
                                ResultBase jobOutput = (ResultBase) Activator.CreateInstance(this._OutputTypes[outputTypeName]);
                                jobOutput = jobOutput.Instantiate(resultNode);
                                Dictionary<string, Type> readableMembers =
                                    jobOutput.GetBindableMemberNames(null);
                                Connector wfConnector = wfDesign.GetConnector(upstreamComponent, wfComponent);
                                if (wfConnector!=null && wfConnector.FieldMappers != null)
                                {
                                    foreach (CodeFieldMapper fieldMapper in wfConnector.FieldMappers)
                                    {
                                        if (readableMembers.ContainsKey(fieldMapper.SrcFieldName) &&
                                            writableInstructionFields.ContainsKey(fieldMapper.TgtFieldName) &&
                                            wfConnector.FromComponentID == fieldMapper.SrcComponentID &&
                                            wfComponent.ComponentID == fieldMapper.TgtComponentID)
                                        {
                                            // MemberInfo outboundField = writableMembers[fieldMapper.TgtFieldName];
                                            object inboundFieldValue =
                                                jobOutput.ReadMemberValue(fieldMapper.SrcFieldName);
                                            if (inboundFieldValue != null)
                                            {
                                                if (!string.IsNullOrEmpty(fieldMapper.AssignmentTransformation))
                                                {

                                                    MethodArgument arg =
                                                        new MethodArgument(
                                                            "fromObj", inboundFieldValue.GetType(), false, inboundFieldValue);
                                                    List<MethodArgument> args = new List<MethodArgument>();
                                                    args.Add(arg);
                                                    Type returnType = typeof (string);
                                                    FieldInfo toField =
                                                        instruction.GetType().GetField(fieldMapper.TgtFieldName);
                                                    if (toField!=null)
                                                    {
                                                        returnType = toField.FieldType;
                                                    }
                                                    else
                                                    {
                                                        PropertyInfo toProp = instruction.GetType().GetProperty(fieldMapper.TgtFieldName);
                                                        if(toProp !=null)
                                                        {
                                                            returnType = toProp.PropertyType;
                                                        }
                                                        // returnType = ((PropertyInfo)outboundField).PropertyType;
                                                    }
                                                    MethodConstructInfo methodBuilder =
                                                        new MethodConstructInfo(returnType, "Convert", args,
                                                                                fieldMapper.AssignmentTransformation);
                                                    object tmp = CodeDomUtil.ExecuteMethod(methodBuilder);
                                                    instruction.SetMemberValue(fieldMapper.TgtFieldName, tmp);
                                                    //if (outboundField.MemberType == MemberTypes.Field)
                                                    //{
                                                    //    ((FieldInfo) outboundField).SetValue(instruction, tmp);
                                                    //}
                                                    //else
                                                    //{
                                                    //    ((PropertyInfo) outboundField).SetValue(instruction, tmp, null);
                                                    //}
                                                }
                                                else
                                                {
                                                    instruction.SetMemberValue(fieldMapper.TgtFieldName, inboundFieldValue);
                                                }
                                                //else if (outboundField.MemberType == MemberTypes.Field)
                                                //{
                                                //    if (inboundFieldValue.GetType() ==
                                                //        ((FieldInfo) outboundField).FieldType)
                                                //    {
                                                //        ((FieldInfo) outboundField).SetValue(instruction,
                                                //                                             inboundFieldValue);
                                                //    }
                                                //    else
                                                //    {
                                                //        object tmp = Convert.ChangeType(inboundFieldValue,
                                                //                                        ((FieldInfo) outboundField).
                                                //                                            FieldType);
                                                //        ((FieldInfo) outboundField).SetValue(instruction, tmp);
                                                //    }
                                                //}
                                                //else
                                                //{
                                                //    if (inboundFieldValue.GetType() ==
                                                //        ((PropertyInfo) outboundField).PropertyType)
                                                //    {
                                                //        ((PropertyInfo) outboundField).SetValue(instruction,
                                                //                                                inboundFieldValue, null);
                                                //    }
                                                //    else
                                                //    {
                                                //        object tmp = Convert.ChangeType(inboundFieldValue,
                                                //                                        ((PropertyInfo) outboundField).
                                                //                                            PropertyType);
                                                //        ((PropertyInfo) outboundField).SetValue(instruction, tmp, null);
                                                //    }
                                                //}
                                            }
                                        }
                                    }
                                }
                            }
                        }
                        
                    }
                }
                return instruction;
            }
            else
            {
                return null;
            }
        }

        
        #endregion

        #region notification
        private void LogErrorMessage(string msg)
        {
            if (msg.Length > 3990)
                msg = msg.Substring(0, 3990) + "...";
            this._Log.WriteEntry(msg, EventLogEntryType.Error);
        }

        private void LogInfo(string msg)
        {
            if (msg.Length > 3990)
                msg = msg.Substring(0, 3990) + "...";
            this._Log.WriteEntry(msg, EventLogEntryType.Information);
        }

        private void SendNotification(
            string smtpServer, bool enableSSL,
            bool requireAuthentication, string userID, string password, 
            string fromAddr,
            string[] toAddresses, string[] ccAddresses,
            string subj, string body)
        {
            try
            {
                MailMessage message = new MailMessage();
                message.From = new MailAddress(fromAddr);
                foreach (string toAddr in toAddresses)
                {
                    message.To.Add(new MailAddress(toAddr));
                }
                if (ccAddresses != null && ccAddresses.Length > 0)
                {
                    foreach (string ccAddr in ccAddresses)
                    {
                        message.CC.Add(new MailAddress(ccAddr));
                    }
                }
                message.Subject = subj;
                message.Body = body;
                message.Priority = MailPriority.High;
                message.IsBodyHtml = true;
                SmtpClient smtp = new SmtpClient(smtpServer);
                if (requireAuthentication)
                {
                    smtp.Credentials = new NetworkCredential(userID, password);
                }
                if (enableSSL)
                {
                    smtp.EnableSsl = true;
                }
                smtp.SendAsync(message, null);
            }
            catch (Exception ex)
            {
                this.LogErrorMessage(ex.ToString());
                this.LogErrorMessage(ex.StackTrace);
            }
        }
        #endregion
    }
}
